package com.intct.flink.watermark;

import com.intct.hbase.bean.WindowBean;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author gufg
 * @since 2025-06-30 10:11
 */
public class WatermarkMonotoDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度
        env.setParallelism(1);

        // 数据源
        DataStreamSource<String> sourceDS = env.socketTextStream("jd-node", 8888);

        // 转换操作
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS.map(source -> {
            String[] sources = source.split(",");
            return WindowBean.builder().name(sources[0]).ts(Long.valueOf(sources[1])).vc(Integer.valueOf(sources[2])).build();
        });

        // 定义watermark策略
        WatermarkStrategy<WindowBean> watermarkStrategy = WatermarkStrategy
                // 指定watermark生成有序
                .<WindowBean>forMonotonousTimestamps()
                // 指定所使用的时间戳：EventTime
                .withTimestampAssigner(new SerializableTimestampAssigner<WindowBean>() {
                    @Override
                    public long extractTimestamp(WindowBean element, long recordTimestamp) {
                        System.out.println("recordTimestamp: " + recordTimestamp);
                        return element.getTs() * 1000L;
                    }
                });

        // 指定watermark
        SingleOutputStreamOperator<WindowBean> sourceWatermarksDS = mapDS.assignTimestampsAndWatermarks(watermarkStrategy);

        // 分区
        KeyedStream<WindowBean, String> keyByDS = sourceWatermarksDS.keyBy(key -> key.getName());

        // 窗口
        WindowedStream<WindowBean, String, TimeWindow> windowDS = keyByDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<String> processDS = windowDS.process(new ProcessWindowFunction<WindowBean, String, String, TimeWindow>() {
            @Override
            public void process(String key, ProcessWindowFunction<WindowBean, String, String, TimeWindow>.Context context,
                                Iterable<WindowBean> elements, Collector<String> out) throws Exception {

                long count = elements.spliterator().estimateSize();

                long startTs = context.window().getStart();
                String startTime = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss");
                long endTs = context.window().getEnd();
                String endTime = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss");

                out.collect("key=" + key + "窗口[开始时间:" + startTime + " 结束时间: " + endTime + "] ;数据的条数=" + count + "数据=" + elements.toString());

            }
        });

        // 输出
        processDS.print();

        // 启动作业
        env.execute();
    }
}
